package com.example.java.time;

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import javax.annotation.Nullable;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;

// 数据进入哪个窗口还是看eventtime，watermark只是为了触发窗口执行的而已
// 以eventtime处理数据时，窗口划分不是以进入的第一数据为开始，而是会根据初始的EventTime时间每隔一段时间划分一个窗口
public class WatermarkDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 指定按eventtime对数据进行处理
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        // 设定水印生成周期
        env.getConfig().setAutoWatermarkInterval(1000L);
        DataStreamSource<String> data = env.socketTextStream("linux121", 8888);
        SingleOutputStreamOperator<Tuple2<String, Long>> maped = data
                .map(new MapFunction<String, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new Tuple2<>(split[0], Long.valueOf(split[1]));
                    }
                });

        // 为maped后的数据生成水印
        SingleOutputStreamOperator<Tuple2<String, Long>> watermarksed = maped.assignTimestampsAndWatermarks(
                new WatermarkStrategy<Tuple2<String, Long>>() {
                    @Override
                    public WatermarkGenerator<Tuple2<String, Long>> createWatermarkGenerator(
                            WatermarkGeneratorSupplier.Context context) {
                        return new WatermarkGenerator<Tuple2<String, Long>>() {
                            private Long currentMaxTimestamp = Long.MIN_VALUE;
                            private SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-DD HH:mm:ss.SSS");

                            // 每一个event发送过来时就会调用一次
                            @Override
                            public void onEvent(Tuple2<String, Long> event, long eventTimestamp,
                                                WatermarkOutput output) {
                                // 比较发送过来的数据，拿到当前接收到的数据中最大时间戳的数据
                                currentMaxTimestamp = Math.max(currentMaxTimestamp, event.f1);
                                System.out.println("currentMaxTimestamp:" + currentMaxTimestamp
                                        + " formatted: " + sdf.format(currentMaxTimestamp));
                            }

                            // 周期性(默认200ms)的生成watermark并放射出去(广播方式,因为数据量大,没必要一条一条生成watermark)
                            @Override
                            public void onPeriodicEmit(WatermarkOutput output) {
                                System.out.println("onPeriodicEmit");
                                long maxOutOfOrderness = 3000;
                                output.emitWatermark(new Watermark(currentMaxTimestamp - maxOutOfOrderness));

                            }
                        };
                    }
                    // 指明根据哪个字段生成水印(即eventtime字段是哪个)
                }.withTimestampAssigner(((element, recordTimestamp) -> element.f1)));

        // 老版本生成水印方式
//        SingleOutputStreamOperator<Tuple2<String, Long>> oldWatermarksed = maped
//                .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Long>>() {
//                    Long currentMaxTimestamp = 0l;
//                    final Long maxOutOfOrderness = 10000l;
//                    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
//
//                    @Nullable
//                    @Override
//                    public org.apache.flink.streaming.api.watermark.Watermark getCurrentWatermark() {
//                        return new org.apache.flink.streaming.api.watermark.Watermark(
//                                currentMaxTimestamp - maxOutOfOrderness);
//                    }
//
//                    @Override
//                    public long extractTimestamp(Tuple2<String, Long> element, long previousElementTimestamp) {
//                        long timestamp = element.f1;
//                        currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
//                        System.out.println("key:" + element.f0
//                                        + "...eventtime:[" + element.f1 + "|" + sdf.format(element.f1)
//                                /*+ "],currentMaxTimestamp:[" + currentMaxTimestamp + "|" + sdf.format(currentMaxTimestamp)*/
//                                /*+ "],watermark:[" + getCurrentWatermark().getTimestamp() + "| " + sdf.format(getCurrentWatermark().getTimestamp() + "]")*/);
//                        System.out.println("currentMaxTimestamp" + currentMaxTimestamp + "..." + sdf.format(currentMaxTimestamp));
//                        System.out.println("watermark:" + getCurrentWatermark().getTimestamp() + "..." + sdf.format(getCurrentWatermark().getTimestamp()));
//                        return timestamp;
//                    }
//                });

        // 对生成水印的数据进行分组聚合,然后进行window操作
        SingleOutputStreamOperator<String> res = watermarksed.keyBy(value -> value.f0)
                .window(TumblingEventTimeWindows.of(Time.seconds(3)))
                .apply(new WindowFunction<Tuple2<String, Long>, String, String, TimeWindow>() {
                    @Override
                    public void apply(String key, TimeWindow window, Iterable<Tuple2<String, Long>> input,
                                      Collector<String> out)
                            throws Exception {
                        ArrayList<Long> list = new ArrayList<>();
                        Iterator<Tuple2<String, Long>> it = input.iterator();
                        while (it.hasNext()) {
                            Tuple2<String, Long> next = it.next();
                            list.add(next.f1);
                        }
                        Collections.sort(list);
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
                        String result = key + "," + list.size() + "," + sdf.format(list.get(0)) + ","
                                + sdf.format(list.get(list.size() - 1)) + ","
                                + sdf.format(window.getStart()) + "," + sdf.format(window.getEnd());
                        out.collect(result);
                    }
                });

        res.print();
        env.execute();
    }
}

